package io.a.consumer;

import org.apache.pulsar.client.api.*;

public class PulsarConsumer {


    private static final String localClusterUrl = "pulsar://pulsar-d1-dev.guideirai.online:26650";


    public static void main(String[] args) {
        try {
            Consumer<byte[]> consumer = getConsumer();

            while (true) {
                Message<byte[]> receive = consumer.receive();
                System.out.println("received : " + new String(receive.getData()));
                consumer.acknowledge(receive);
            }
        } catch (PulsarClientException e) {
            throw new RuntimeException(e);
        }
    }


    private static Consumer<byte[]> getConsumer() throws PulsarClientException {
        PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(localClusterUrl).build();

        return pulsarClient.newConsumer()
                .topic("data_sync_ai_platform_event")
                .subscriptionName("ai-subscription")
                .subscriptionType(SubscriptionType.Failover)
                .subscribe();
    }


}
